package com.jackrain.nea.clould.mq;

import com.alibaba.dubbo.common.utils.NetUtils;
import com.aliyun.openservices.ons.api.Action;
import com.aliyun.openservices.ons.api.ConsumeContext;
import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.impl.util.MsgConvertUtil;
import com.jackrain.nea.clould.dao.ONSConsumerMQDao;
import com.jackrain.nea.clould.dao.ONSSendMontitorMQBodyDao;
import com.jackrain.nea.clould.dao.SendMonitorDao;
import com.jackrain.nea.clould.dao.TopicDao;
import com.jackrain.nea.clould.model.ONSConsumerMQModel;
import com.jackrain.nea.clould.model.ONSSendMonitorMQBodyModel;
import com.jackrain.nea.clould.model.SendMonitorModel;
import com.jackrain.nea.util.Tools;
import com.jackrain.nea.util.ValueHolder;
import lombok.extern.slf4j.Slf4j;
import org.springframework.cglib.proxy.MethodInterceptor;
import org.springframework.cglib.proxy.MethodProxy;

import java.io.IOException;
import java.io.PrintWriter;
import java.io.Serializable;
import java.io.StringWriter;
import java.lang.reflect.Method;

/**
 * Created by ZC 2017/5/31.
 */
@Slf4j
public class MQMessageLogIntercept implements MethodInterceptor {

    private TopicDao topicDao = new TopicDao();
    private SendMonitorDao sendMonitorDao = new SendMonitorDao();
    private ONSConsumerMQDao onsConsumerMQDao = new ONSConsumerMQDao();
    private ONSSendMontitorMQBodyDao onsSendMontitorMQBodyDao = new ONSSendMontitorMQBodyDao();

    private long onsConsumerMQ_id;
    private long sendMonitorID = 0;

    @Override
    public Object intercept(Object o, Method method, Object[] objects, MethodProxy methodProxy) throws Throwable {

        //先执行consume,再执行这个mqConsume,最后执行consume完毕
        Object rev = null;
        try {
            //执行织入的日志
            if (method.getName().equals("consume")) {
                //1
                beforeConsume((Message) objects[0], (ConsumeContext) objects[1]);
                //2
                rev = methodProxy.invokeSuper(o, objects);
                //5
            } else if (method.getName().equals("mqConsume")) {
                //3
                rev = methodProxy.invokeSuper(o, objects);
                //4
                AfterConsume((ValueHolder) rev, (Message) objects[0], (ConsumeContext) objects[1]);
            }
        } catch (Exception e) {
            //更新错误日志
            StringWriter sw = new StringWriter();
            PrintWriter pw = new PrintWriter(sw);
            e.printStackTrace(pw);
            ThrowableConsume((Message) objects[0], (ConsumeContext) objects[1], sw.toString());
        }

        return rev;
    }

    /**
     * MQ日志
     */
    public void beforeConsume(Message message, ConsumeContext context) {

        //保存发送消息信息
        try {
            //通过key查找发送消息ID
            sendMonitorID = sendMonitorDao.queryID(message.getKey());
            log.debug("beforeConsume id:" + sendMonitorID);
            if (sendMonitorID <= 0) {
                //新增发送消息信息
                int oms_topicid = topicDao.queryTopicID(message.getTopic());
                SendMonitorModel sendMonitorModel = new SendMonitorModel();
                sendMonitorModel.setPtype("1");//1:线上数据,2:线下数据,3:线下消费回执,4:消息监控提醒
                sendMonitorModel.setTopicid(oms_topicid);
                sendMonitorModel.setTag(message.getTag());
                sendMonitorModel.setMsgkey(message.getKey());
                sendMonitorModel.setState("1");//0:待发送,1:待消费,2:消费成功,
                sendMonitorModel.setConsumer_error("Y");//消费是否存在异常
                sendMonitorDao.insert(sendMonitorModel);
                sendMonitorID = sendMonitorModel.getId();
            }
        } catch (Exception e) {
            e.printStackTrace();
        }

        //保存消费信息
        ONSConsumerMQModel onsConsumerMQModel = new ONSConsumerMQModel();
        onsConsumerMQModel.setConsumerip(NetUtils.getLocalAddress().getHostAddress());
        onsConsumerMQModel.setMsgkey(message.getKey());
        onsConsumerMQModel.setOns_xtype("1");//云端数据消费
        onsConsumerMQModel.setState("0");//未消费
        onsConsumerMQModel.setOns_sendmonitorid(sendMonitorID);
        onsConsumerMQModel.setAd_client_id(37);
        try {
            onsConsumerMQDao.insert(onsConsumerMQModel);
            onsConsumerMQ_id = onsConsumerMQModel.getId();
        } catch (Exception e) {
            e.printStackTrace();
        }


    }

    /**
     * MQ日志
     */
    public void AfterConsume(ValueHolder valueHolder, Message message, ConsumeContext context) {


        if (valueHolder != null) {
            int totCount = Tools.getInt(valueHolder.get("totCount"), 0);
            int successCount = Tools.getInt(valueHolder.get("successCount"), 0);
            //更新数量
            sendMonitorDao.updateDataCount(totCount, successCount, message.getKey());
            //是否成功
            if (Tools.getInt(valueHolder.get("code"), -1) == 0) {//成功
                onsConsumerMQDao.update("2", "", "", successCount, onsConsumerMQ_id);
                //成功回写sendMonitor
                sendMonitorDao.updateStateIsErr("2", "N", message.getKey());
            } else {
                String error = String.valueOf(valueHolder.get("errorMsg"));
                String des = "";
                try {
                    Serializable ser = MsgConvertUtil.objectDeserialize(message.getBody());
                    des = ser.toString();
                } catch (IOException e) {
                    e.printStackTrace();
                } catch (ClassNotFoundException e) {
                    e.printStackTrace();
                }
                if (successCount > 0) {//部分成功
                    onsConsumerMQDao.update("3", error, des, successCount, onsConsumerMQ_id);
                    //成功回写sendMonitor
                    sendMonitorDao.updateStateIsErr("2", "N", message.getKey());
                } else {//全部失败
                    onsConsumerMQDao.update("1", error, des, successCount, onsConsumerMQ_id);
                    //失败回写sendMonitor
                    sendMonitorDao.updateStateIsErr("1", "Y", message.getKey());
                }
            }
            //记录库存数量
            String qtySumTotal = String.valueOf(valueHolder.get("qtySumTotal"));
            String qtySumSuccess =String.valueOf(valueHolder.get("qtySumSuccess"));
            String qtySumTimeOut = String.valueOf(valueHolder.get("qtySumTimeOut"));
            ONSSendMonitorMQBodyModel onsSendMonitorMQBodyModel = new ONSSendMonitorMQBodyModel();
            onsSendMonitorMQBodyModel.setOns_sendmonitor_id(sendMonitorID);
            onsSendMonitorMQBodyModel.setAbstract1(qtySumTotal);
            onsSendMonitorMQBodyModel.setAbstract2(qtySumSuccess);
            onsSendMonitorMQBodyModel.setAbstract3(qtySumTimeOut);
            onsSendMonitorMQBodyModel.setMsgKey(message.getKey());
            onsSendMontitorMQBodyDao.insert(onsSendMonitorMQBodyModel);
        }
    }


    public void ThrowableConsume(Message message, ConsumeContext context, String error) {

        //失败回写sendMonitor
        sendMonitorDao.updateStateIsErr("1", "Y", message.getKey());

        String des = "";
        try {
            Serializable ser = MsgConvertUtil.objectDeserialize(message.getBody());
            des = ser.toString();
        } catch (IOException e) {
            e.printStackTrace();
        } catch (ClassNotFoundException e) {
            e.printStackTrace();
        }

        //回写消费错误
        onsConsumerMQDao.update("1", error, des, 0, onsConsumerMQ_id);

    }


}
